package com.bff.gaia.mix.api.xjoin.client.hbase.async.rowkeydealer;

import com.bff.gaia.mix.api.functions.async.ResultFuture;
import com.bff.gaia.mix.api.xjoin.core.enums.ECacheContentType;
import com.bff.gaia.mix.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.mix.api.xjoin.core.side.CacheMissVal;
import com.bff.gaia.mix.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.mix.api.xjoin.core.side.SideInfo;
import com.bff.gaia.mix.api.xjoin.core.side.cache.AbsSideCache;
import com.bff.gaia.mix.api.xjoin.core.side.cache.CacheObj;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;
import com.bff.gaia.types.Row;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;

public class RowKeyEqualModeDealer extends AbsRowKeyModeDealer {
	private static final Logger LOG = LoggerFactory.getLogger(RowKeyEqualModeDealer.class);

	public RowKeyEqualModeDealer(Map<String, String> colRefType, String[] colNames, HTable hBaseClient,
	                             boolean openCache, SideInfo sideInfo) {
		super(colRefType, colNames, hBaseClient, openCache, sideInfo);
	}


	@Override
	public void asyncGetData(String tableName, String rowKeyStr, Row input, ResultFuture<Row> resultFuture,
	                         AbsSideCache sideCache) {

		try {
			Result result = hBaseClient.get(createGet(rowKeyStr));
			Map<String, Object> sideMap = getResultMap(result);
			if (sideMap.isEmpty()) {
				dealMissKey(input, resultFuture);

				if (openCache) {
					sideCache.putCache(rowKeyStr, CacheMissVal.getMissKeyObj());
				}
			} else {
				List<Object> sideVal = Lists.newArrayList();
				for (String key : colNames) {
					Object val = sideMap.get(key);
					if (val == null) {
						System.out.println("can't get data with column " + key);
						LOG.error("can't get data with column " + key);
					}
					sideVal.add(val);
				}
				Row row = fillData(input, sideVal);
				if (openCache) {
					sideCache.putCache(rowKeyStr, CacheObj.buildCacheObj(ECacheContentType.SingleLine, row));
				}

				resultFuture.complete(Collections.singleton(Row.project(row, outFieldIndexs)));
			}
		} catch (IOException e) {
			LOG.error("record:" + input);
			LOG.error("get side record exception:" + "");
			resultFuture.complete(null);
		}
	}
}